package main
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.FileSystem
import org.apache.hadoop.fs.Path
import org.apache.hadoop.fs.FSDataInputStream
import org.apache.hadoop.io.IOUtils
import org.apache.hadoop.hdfs.DistributedFileSystem
import java.sql.Timestamp

object HDFSUtil {
  private val fs: FileSystem get() {
    val conf = Configuration()
    conf.set("fs.defaultFS", "hdfs://192.168.244.143:9000")
    return FileSystem.get(conf)
  }
  fun createFile(path: String, contents: ByteArray): Unit {
    val dst = Path(path)
    val outStream = fs.create(dst)
    outStream.write(contents)
    outStream.close()
    println("文件创建成功")
  }

  fun close() {
    fs.close()
  }

  fun createFile(path: String, contents: String): Unit {
    createFile(path, contents.toByteArray())
  }

  fun deleteFile(path: String): Unit {
    val dst = Path(path)
    val isOk = fs.deleteOnExit(dst)
    println(if (isOk) "删除成功" else "删除失败")
  }

  fun uploadFile(src: String, dst: String): Unit {
    val srcPath = Path(src)
    val dstPath = Path(dst)
    fs.copyFromLocalFile(false, srcPath, dstPath)
    fs.listStatus(dstPath).forEach { file -> println("目标路径为" + file.getPath()) }
  }

  fun readFile(filePath: String): Unit {
    val path = Path(filePath)
    var inputStream: FSDataInputStream? = null
    try {
      inputStream = fs.open(path)
      IOUtils.copyBytes(inputStream, System.out, 4096, false)
    } finally {
      IOUtils.closeStream(inputStream)
    }
  }

  fun showFile(filePath: String, isDetail: Boolean): Unit {
    val path = Path(filePath)
    val dirStatus = fs.getFileStatus(path)
    if (dirStatus.isDirectory) {
      println("这是一个文件夹")
      fs.listStatus(path).forEach { file ->
        if (isDetail) { // hadoop fs -ls -l <filePath>
          println(
            "${if (file.isDirectory) "d" else "-"} \t ${file.owner} \t ${file.group} \t ${file.len} ${
              Timestamp(file.modificationTime)
            } \t ${file.path}"
          )
        } else { // hadoop fs -ls <filePath>
          println(file.path)
        }
      }
    }
  }
    fun showLocation(filePath: String): Unit {
     val conf = Configuration()
      val fs = FileSystem.get(conf)
      val path = Path(filePath)
      val fileStatus = fs.getFileStatus(path)
      val locations = fs.getFileBlockLocations(fileStatus, 0, fileStatus.len)
      locations.forEach { location ->
        location.hosts.forEach { host -> println("host: $host") }
      }
    }

    fun showDataNodeName(): Unit {
      val hdfs: DistributedFileSystem = fs as DistributedFileSystem
      hdfs.dataNodeStats.forEach { dataNode ->
        println("node name: ${dataNode.hostName}")
        println("node ip: ${dataNode.ipAddr}")
        println("node ipc port: ${dataNode.ipcPort}")
      }
    }
  }
